package cc.verywell.pureblock.net;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.util.EncryptUtil;
import cc.verywell.pureblock.util.StringUtil;

/**
* 
* Copyright 2019 TanYaqiu
* @author TanYaqiu 
* @address flat 601, teachers' dormitory building, w, SCNU
* @date 2019年3月15日 上午12:59:21
*/
@Deprecated
public class AioClientImpl0 implements Closeable, ConnectionFreeListener, AioClientInterface<NormalSession, AioClientConnector> {

	final private AioProvider provider;
	public AioClientImpl0(AioProvider provider) {
		this.provider = provider;
		this.provider.addConnectionFreeListener(this);
	}
	
	
	///TODO 应该添加连接队列，当provider.isTooManyPendingConnections()有空位时再连上;
	@Override
	public <A> Future<NormalSession> connect(AioClientConnector connector, A t,CompletionHandler<NormalSession, A> handler) {
		///TODO 留个接口，以后要改成队列
		boolean ok = false;
		if(!ok) {
			NormalSession session = connector.getSession();
			if(session!=null && !session.isClosed()) {
				handler.completed(session, t);
				ok = true;
				if(Pblog.IO)Pblog.IO(connector.getRemote()," is connected at ",session);
			}
		}
		if(!ok) {
			for (String remotestr : connector.getRemote()) {
				InetSocketAddress address = StringUtil.stringToPort(remotestr);
				try {
					Future<NormalSession> to = this.connect(address);
					if(to==null)continue;
					NormalSession n = to.get();//TODO//20200206这里会锁死一条线程
					handler.completed(n, t);
					ok = true;
					return to;
					//break;
				} catch (IOException | InterruptedException | ExecutionException e) {
					if(Pblog.IO)Pblog.IO(e);
				}
			}
		}
		if(!ok) {
			handler.failed(null, t);
		}
		if(!provider.isTooManyPendingConnections()) {
			//TODO 下一个队列
		}
		return AioClientInner.canceledFuture;
	}

	
	@Override
	public void connectionfree() {
		
		// TODO 继续一个队列;
		
	}
	@Override
	public void close() {
		provider.removeConnectionFreeListener(this);
		// TODO 关闭连接队列
		
	}
	
	///////////////////////黑名单类
	private Map<SocketAddress, Long> connectedMap = new ConcurrentHashMap<SocketAddress, Long>();
	private AtomicLong connectedMin = new AtomicLong(0);
	private boolean isConnected(SocketAddress sa) {
		final long TIMEOUT = AioProvider.RECONNECT_TIME_OUT;
		long time = System.currentTimeMillis();
		long last = connectedMin.get()-time;
		if(last>0 && last<TIMEOUT) {
			//if(Pblog.DEBUG)Pblog.DEBUG("isConected normal ",time," remote: ",sa," last: ",last," timeout: ",TIMEOUT);
			return null!=connectedMap.putIfAbsent(sa, time+TIMEOUT);
		}
		synchronized (connectedMap) {
			last = connectedMin.get();
			if(last==0) {//还没有记录
				//if(Pblog.DEBUG)Pblog.DEBUG("isConected first ",time," remote: ",sa," timeout: ",TIMEOUT);
				time += TIMEOUT;
				connectedMap.put(sa,time);
				connectedMin.set(time);
				return false;
			}
			if(time<last) {
				//if(Pblog.DEBUG)Pblog.DEBUG("isConected timechanged ",time," remote: ",sa," last: ",last," timeout: ",TIMEOUT);
				time += TIMEOUT;
				if(time<last-TIMEOUT) {//时钟发生更改
					connectedMap.clear();
					connectedMap.put(sa,time);
					connectedMin.set(time);
					return false;
				}
				return null!=connectedMap.putIfAbsent(sa, time);
			}
			//else 需要清理已经超时的记录
			/*
			 * 只 putIfAbsnet的情况用removeIf, put情况用remove(key,value)才不会误删
			 * 参考 https://stackoverflow.com/questions/29876083/behavior-of-entryset-removeif-in-concurrenthashmap
			 */
			Long hasOld = connectedMap.putIfAbsent(sa, time+TIMEOUT);
			//if(Pblog.DEBUG)Pblog.DEBUG("isConected overtime ",time," remote: ",sa," last: ",last," timeout: ",TIMEOUT," hasOld: ",hasOld);
			long min = 0;
			Iterator<Map.Entry<SocketAddress, Long>> each = connectedMap.entrySet().iterator();
	        while (each.hasNext()) {
	        	Map.Entry<SocketAddress, Long> e = each.next();
	        	if(e.getValue()>time) {
	        		if(min==0 || min>e.getValue())
	        			min=e.getValue();
	        	}else {
	                each.remove();
	            }
	        }
	        connectedMin.set(min);
			return hasOld!=null && hasOld>time; 
		}
	}
	///////////////////////直连类
	
	private Map<SocketAddress, ClientFuture> connectingMap = new ConcurrentHashMap<SocketAddress, ClientFuture>();

	/**
	 * @param remote
	 * @return null if cannot connect; 
	 * @throws IOException
	 */
	public synchronized Future<NormalSession> connect(SocketAddress remote) throws IOException {
		//TODO 应该选用最后一次连接的acg吗 ?
		return connect(null, null, remote);
	}
	/**
	 * 当 connect().get() 正常返回时，channel可能并不在
	 * @param acg
	 * @param bind
	 * @param remote
	 * @return null if cannot connect; 
	 * @throws IOException
	 */
	public synchronized Future<NormalSession> connect(AsynchronousChannelGroup acg,InetSocketAddress bind,SocketAddress remote) throws IOException {
		if(remote==null)throw new NullPointerException();
		ClientFuture f = new ClientFuture(remote);
		{
			ClientFuture old = connectingMap.putIfAbsent(remote, f);
			if(old!=null)return old;
		}
		if(isConnected(remote)) {
			if(Pblog.IO)Pblog.IO("conected recently[",provider.myAlias(),"->(ready)",remote,"]");
			if(Pblog.DEBUG) {
				Long l = connectedMap.get(remote);
				if(l!=null)
					Pblog.DEBUG("conected recently[",provider.myAlias(),"->(ready)",remote,"]","@rest: ",l-System.currentTimeMillis()," of period: ",AioProvider.RECONNECT_TIME_OUT);
			}
			return null;
		}else {
			//if(Pblog.IO)Pblog.IO("connecting[",provider.myAlias(),"->(ready)",remote,"]",Integer.toHexString(f.hashCode()));
			if(Pblog.IO)Pblog.IO(f);
		}
		synchronized (f) {
			try {
				if(acg==null)acg = provider.getDefaultChannelGroup(0);
				AsynchronousSocketChannel ch = AsynchronousSocketChannel.open(acg);
				if(bind!=null) {
					ch.bind(bind);
				}
				provider.pendingSet.add(ch);
				//ClientFuture cf;
				try {
					ch.setOption(StandardSocketOptions.SO_REUSEADDR, true);
					ch.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
					Future<Void> future = ch.connect(remote);
					future.get(AioProvider.TIME_OUT, TimeUnit.SECONDS);
					
					HandshakeChannelClient hc = handshakeChannels.poll();
					if(hc==null)hc = new HandshakeChannelClient(provider.myAlias()); //这里会throw NoSuchAlgorithmException
					hc.init(ch, AioProvider.TIME_OUT, f.setChannel(ch));
					ClientFuture rf = f;
					f = null;
					return rf;
				} catch (Throwable e) {
					if(Pblog.IO)Pblog.IO(e," to ",remote);
					try {
						ch.close();
					} catch (IOException e1) {
						if(Pblog.IO)Pblog.IO(e1);
					}
					provider.pendingSet.remove(ch);
					//e.printStackTrace();
					return null;
				}
			}finally {
				if(null!=f) {
					f.cancel(false);
				}
			}
		}
	}
	
	private class ClientFuture implements Future<NormalSession>,HandshakeFunction<HandshakeChannelClient>,NormalFunction{
		private AsynchronousSocketChannel channel;
		private NormalSession ns;
		private boolean isCancelled,isDone;
		final SocketAddress remote;
		ClientFuture(SocketAddress remote){
			this.remote = remote;
		}
		
		ClientFuture setChannel(AsynchronousSocketChannel ch){
			this.channel = ch;
			return this;
		}
		@Override
		synchronized public boolean cancel(boolean mayInterruptIfRunning) {
			if(isCancelled || isDone)return false;
			isCancelled = isDone = true;
			connectingMap.remove(remote, this);
			if(null!=channel && channel.isOpen()) {
				try {
					channel.shutdownInput();
				} catch (IOException e) {
					if(Pblog.IO)Pblog.IO(this,"@closing: ",e);
				}
				try {
					channel.shutdownOutput();
				} catch (IOException e) {
					if(Pblog.IO)Pblog.IO(this,"@closing: ",e);
				}
				try {
					channel.close();
				} catch (IOException e) {
					if(Pblog.IO)Pblog.IO(this,"@closing: ",e);
				}
			}
			this.notifyAll();
			return true;
		}

		@Override
		public boolean isCancelled() {
			return isCancelled;
		}

		@Override
		public boolean isDone() {
			return isDone;
		}

		@Override
		synchronized public NormalSession get() throws InterruptedException, ExecutionException {
			if(isDone)return ns;
			wait();
			return ns;
		}

		@Override
		synchronized public NormalSession get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
			if(isDone)return ns;
			unit.timedWait(this, timeout);
			return ns;
		}
		
		@Override
		public void handshakeCompleted(HandshakeChannelClient obj, EncryptUtil eu, ByteBuffer unread, int salt) {
			try {
				NormalChannel nc = provider.handshakeCompleted(obj.getChannel(), eu, unread, this, salt, true);
				if(nc==null) { ////致命错误;
					synchronized (this) {
						isCancelled = isDone = true;
						connectingMap.remove(remote, this);
						this.notifyAll();
					}
				}
			}finally {
				handshakeChannels.offer(obj.clearUser());
			}
			
//			synchronized (this) {
//				this.ns = nc.getSession();
//				isDone = true;
//				this.notifyAll();
//			}
//			return nc;
		}
		@Override
		public void handshakeFailed(HandshakeChannelClient obj, Throwable exp) {
			try {
				provider.handshakeFaild(obj.getChannel(), exp);
				synchronized (this) {
					isCancelled = isDone = true;
					connectingMap.remove(remote, this);
					this.notifyAll();
				}
			}finally {
				handshakeChannels.offer(obj.clearUser());
			}
		}
		
		NormalFunction nextfunction;
		@Override
		public void addNext(NormalFunction function) {
			this.nextfunction = function;
		}
		@Override
		public void init(NormalSession channel) {
			if(nextfunction!=null)nextfunction.init(channel);
			synchronized (this) {
				this.ns = channel;
				isDone = true;
				connectingMap.remove(remote, this);
				this.notifyAll();
			}
		}
		@Override
		public void closed(NormalChannel channel) {
			if(nextfunction!=null)nextfunction.closed(channel);
			synchronized (this) {
				isCancelled = isDone = true;
				connectingMap.remove(remote, this);
				this.notifyAll();
			}
		}
		
		@Override
		public String toString() {
			StringBuffer sb = new StringBuffer();
			sb.append("connecting").append("[").append(provider.myAlias()).append("->");
			AsynchronousSocketChannel channel = this.channel;
			if(channel!=null) {
				sb.append("(").append(Integer.toHexString(channel.hashCode())).append(")");
				try {
					sb.append(channel.getRemoteAddress() );
				} catch (IOException e) {
					sb.append(remote).append("<").append(e.getLocalizedMessage()).append(">");
				}
			}else{
				sb.append("(ready)").append(remote);
			}
			sb.append("]").append(Integer.toHexString(hashCode()));
			return sb.toString();
		}
		
	}
	
	private BlockingQueue<HandshakeChannelClient> handshakeChannels = new LinkedBlockingQueue<>();
//	private HandshakeFunction<HandshakeChannelClient> acceptHandshake = new HandshakeFunction<HandshakeChannelClient>() {
//		@Override
//		public void completed(HandshakeChannelClient obj, AsynchronousSocketChannel channel, EncryptUtil eu, ByteBuffer unread) {
//			handshakeChannels.offer(obj.clearUser());
//			NormalChannel nc = provider.handshakeCompleted(channel, eu, unread, null);
//		}
//		@Override
//		public void failed(HandshakeChannelClient obj, Throwable exp) {
//			handshakeChannels.offer(obj.clearUser());
//			provider.handshakeFaild(obj, exp);
//		}
//	};


}
